package architect;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import static java.lang.System.err;

/**
 * @description: Flink快速启动
 * @author: guanxin
 * @date: 2018/10/21 上午10:59
 * @since: JDK 1.8
 */
public class SocketTextStreamWordCount {

    public static void main(String[] args) throws Exception {
        // 参数检查
        if (args.length != 2) {
            err.println("USAGE:\nSocketTextStreamWordCount <hostname> <port>");
            return;
        }

        String hostName = args[0];
        int port = Integer.parseInt(args[1]);

        //设置环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder();

        //获取数据
        DataStream<String> text = env.socketTextStream(hostName, port);

        //计数
        DataStream<Tuple2<String, Integer>> counts = text.flatMap(new LineSplitter())
                .keyBy((KeySelector<Tuple2<String, Integer>, Object>) stringIntegerTuple2 -> "0")
                .sum(1);
        counts.print();

        env.execute("Java WordCount from SocketTextStream Example");
    }

    public static final class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) {
            String[] tokens = s.toLowerCase().split("\\W+");

            for (String token: tokens) {
                if (token.length() > 0) {
                    collector.collect(new Tuple2<>(token, 1));
                }
            }
        }
    }
}
